package com.alarm.ecp;

import com.alarm.ecp.pojo.TransactionEvent;
import com.alarm.ecp.pojo.AlertEvent;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.functions.PatternProcessFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.util.List;
import java.util.Map;

/**
 * 找出那些 24 小时内至少 5 次有效交易的账户
 */
public class TransactionAlarmFlink {
    private static final Logger logger = LoggerFactory.getLogger(TransactionAlarmFlink.class);

    public static void main(String[] args) throws Exception {
        // 设置匹配
        Pattern<TransactionEvent, TransactionEvent> pattern = Pattern.<TransactionEvent>begin("start").where(
                        new SimpleCondition<TransactionEvent>() {
                            @Override
                            public boolean filter(TransactionEvent transactionEvent) {
                                return transactionEvent.getAmount() > 0;
                            }
                        }
                ).timesOrMore(5)
                .within(Time.hours(24));

        // 初始化环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设值事件时间
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        // 处理数据源（设置水印）
        DataStream<TransactionEvent> source = env.fromElements(
                new TransactionEvent("100XX", 0.0D, 1597905234000L),
                new TransactionEvent("100XX", 100.0D, 1597905235000L),
                new TransactionEvent("100XX", 200.0D, 1597905236000L),
                new TransactionEvent("100XX", 300.0D, 1597905237000L),
                new TransactionEvent("100XX", 400.0D, 1597905238000L),
                new TransactionEvent("100XX", 500.0D, 1597905239000L),
                new TransactionEvent("101XX", 0.0D, 1597905240000L),
                new TransactionEvent("101XX", 100.0D, 1597905241000L)
        ).assignTimestampsAndWatermarks(
                WatermarkStrategy.<TransactionEvent>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        .withTimestampAssigner((event, timestamp) -> event.getTimeStamp())
        ).keyBy((KeySelector<TransactionEvent, Object>) TransactionEvent::getAccout);

        // 绑定时间源和匹配
        PatternStream<TransactionEvent> patternStream = CEP.pattern(source, pattern);

        // 执行匹配输出告警
        SingleOutputStreamOperator<AlertEvent> process = patternStream.process(new PatternProcessFunction<TransactionEvent, AlertEvent>() {
            @Override
            public void processMatch(Map<String, List<TransactionEvent>> match, Context ctx, Collector<AlertEvent> out) throws Exception {
                List<TransactionEvent> start = match.get("start");
                List<TransactionEvent> next = match.get("next");
                System.err.println("start:" + start + ",next:" + next);
                out.collect(new AlertEvent(start.get(0).getAccout(), "连续有效交易！"));
            }
        });

        process.printToErr();
        env.execute("execute cep");
    }
}
